package com.github.xiaolyuh.bean;


import com.github.xiaolyuh.PubSubWithAckRedisManager;
import com.github.xiaolyuh.configcenter.ConfigCenter;
import com.github.xiaolyuh.constant.CacheConstant;
import com.github.xiaolyuh.entity.MqMessageModel;
import com.github.xiaolyuh.enums.MqMessageTypeEnum;
import com.github.xiaolyuh.exception.PromotionMonitorException;
import com.github.xiaolyuh.utils.FastJSONHelper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

public class LocalCacheFlushByMsgBean implements InitializingBean {

    private static final Logger logger= LoggerFactory.getLogger(LocalCacheFlushByMsgBean.class);

    private ExecutorService executorService;

    @Autowired
    private PubSubWithAckRedisManager pubSubWithAckRedisManager;

    @Override
    public void afterPropertiesSet() throws Exception {
        List<String> siteCodes= ConfigCenter.getSiteCodes();
        executorService= Executors.newFixedThreadPool(siteCodes.size(),new ThreadFactoryBuilder().setNameFormat("msg-clear-localcache-pool-%d").build());
        CountDownLatch latch=new CountDownLatch(siteCodes.size());
        for (String siteCode:siteCodes){
            executorService.execute(new SubscribeChannelRunable(siteCode,new ClearLocalCacheJedisPubSub(siteCode,latch)));
        }

        long startNano=System.nanoTime();
        long nanosTimeout= TimeUnit.MINUTES.toNanos(5);
        latch.await(nanosTimeout,TimeUnit.NANOSECONDS);
        if(System.nanoTime()-startNano>=nanosTimeout){
            throw new TimeoutException("订阅本地缓存消息队列超时");
        }
    }

    private class ClearLocalCacheJedisPubSub extends JedisPubSub{
        private String siteCode;
        private CountDownLatch countDownLatch;
        private String ip;

        public ClearLocalCacheJedisPubSub(String siteCode,CountDownLatch countDownLatch) {
            this.siteCode = siteCode;
            ip= IPUtils.getAddress();
            this.countDownLatch=countDownLatch;
        }

        @Override
        public void onMessage(String channel, String message) {
            try{
                logger.info("收到刷新本地缓存消息：{}",message);
                MqMessageModel mqMessage= FastJSONHelper.deserialize(message,MqMessageModel.class);
                if(StringUtils.isEmpty(mqMessage.getContent())){
                    logger.warn("刷新本地缓存收到空消息，确认消息发送是否正常");
                    return;
                }
                MqMessageTypeEnum typeEnum = MqMessageTypeEnum.valueOf(mqMessage.getType());
                if(Objects.isNull(typeEnum)){
                    logger.error("收到不可识别的消息类型。type = {}",mqMessage.getType());
                    return;
                }
                typeEnum.process(mqMessage.getContent());
                if(!pubSubWithAckRedisManager.ackMsg(siteCode,mqMessage.getMsgId(),ip)){
                    String failMsg=String.format("确认消息失败。siteCode= %s msgId= %s ip= %s",siteCode,mqMessage.getMsgId(),ip);
                    logger.error(failMsg, PromotionMonitorException.of(PromotionMonitorException.DEFAULT_CODE,failMsg));
                }
            }catch (Exception e){
                logger.error("消费消息失败：{}",message,e);
            }

        }

        @Override
        public void onSubscribe(String channel, int subscribedChannels) {
            logger.info("subscribe channel {} success",channel);
            if(Objects.nonNull(countDownLatch)){
                countDownLatch.countDown();
            }
        }
    }

    private class SubscribeChannelRunable implements Runnable{
        private final Logger logger=LoggerFactory.getLogger(SubscribeChannelRunable.class);
        private static final int maxReConnectMs=10000;
        private JedisPubSub pubSub;
        private String siteCode;
        private int reConnectMs=500;

        public SubscribeChannelRunable(String siteCode,JedisPubSub pubSub) {
            this.pubSub = pubSub;
            this.siteCode = siteCode;
        }

        @Override
        public void run() {
            while (true){
                try{
                    pubSubWithAckRedisManager.subscribe(siteCode,pubSub, CacheConstant.DEFAULT_CHANNEL);
                }catch (JedisConnectionException jce){
                    logger.error(jce.getMessage(),jce);
                    sleep(reConnectMs);
                }catch (JedisException je){
                    logger.error(je.getMessage(),je);
                    sleep(reConnectMs);
                }catch (Exception e){
                    logger.error(e.getMessage(),e);
                    break;
                }
                int incr=reConnectMs*2;
                reConnectMs=incr>maxReConnectMs?maxReConnectMs:incr;
            }
        }

        private void sleep(int reConnectMs){
            try{
                Thread.sleep(reConnectMs);
            }catch (Exception e){
                logger.error(e.getMessage(),e);
            }
        }
    }
}
